package com.baidu.automata.core.flow.steps;

import org.apache.log4j.Logger;

import com.baidu.automata.core.flow.BaseOpStep;
import com.baidu.automata.core.type.LongField;
import com.baidu.automata.core.type.StringField;
import com.baidu.cleaner.TextFileCleaner;

public class CleanFileStep extends BaseOpStep {
	private final static Logger logger = Logger.getLogger("flow_Platform");

	private StringField inputFieldSeparator = null;
	private StringField inputFilePath = null;
	private StringField inputFileEncoding = null;
	private StringField outputFieldSeparator = null;
	private StringField outputFilePath = null;
	private StringField outputFileEncoding = null;
	private StringField logFilePath = null;
	private StringField cleanerId = null;

	public StringField getInputFieldSeparator() {
		return inputFieldSeparator;
	}

	public void setInputFieldSeparator(StringField inputFieldSeparator) {
		this.inputFieldSeparator = inputFieldSeparator;
	}

	public StringField getInputFilePath() {
		return inputFilePath;
	}

	public void setInputFilePath(StringField inputFilePath) {
		this.inputFilePath = inputFilePath;
	}

	public StringField getOutputFieldSeparator() {
		return outputFieldSeparator;
	}

	public void setOutputFieldSeparator(StringField outputFieldSeparator) {
		this.outputFieldSeparator = outputFieldSeparator;
	}

	public StringField getOutputFilePath() {
		return outputFilePath;
	}

	public void setOutputFilePath(StringField outputFilePath) {
		this.outputFilePath = outputFilePath;
	}

	public StringField getOutputFileEncoding() {
		return outputFileEncoding;
	}

	public void setOutputFileEncoding(StringField outputFileEncoding) {
		this.outputFileEncoding = outputFileEncoding;
	}

	public StringField getInputFileEncoding() {
		return inputFileEncoding;
	}

	public void setInputFileEncoding(StringField inputFileEncoding) {
		this.inputFileEncoding = inputFileEncoding;
	}

	public StringField getLogFilePath() {
		return logFilePath;
	}

	public void setLogFilePath(StringField logFilePath) {
		this.logFilePath = logFilePath;
	}

	public StringField getCleanerId() {
		return cleanerId;
	}

	public void setCleanerId(StringField cleanerId) {
		this.cleanerId = cleanerId;
	}

	@Override
	protected int execute() {
		TextFileCleaner p = null;
		try {
			p = new TextFileCleaner(this.inputFilePath.getValue(), this.inputFileEncoding.getValue(), this.inputFieldSeparator.getValue(), this.outputFilePath.getValue(),
					this.outputFileEncoding.getValue(), this.outputFieldSeparator.getValue(), this.logFilePath.getValue(), this.cleanerId.getValue());
			logger.info("start cleaning the file...");
			logger.info(String.format("src file name: %s", this.inputFilePath.getValue()));
			logger.info(String.format("src file encoding: %s", this.inputFileEncoding.getValue()));
			logger.info(String.format("dest file name: %s", this.outputFilePath.getValue()));
			logger.info(String.format("dest file encoding: %s", this.outputFileEncoding.getValue()));
			logger.info(String.format("log file name: %s", this.logFilePath.getValue()));
			logger.info(String.format("cleaner id: %s", this.cleanerId.getValue()));

			p.parse();

			logger.info("cleaning file finished");

			LongField rowNumField = new LongField("row_num");
			rowNumField.setValue(Long.valueOf(p.getRowNum()));
			this.addDataField(rowNumField);
			logger.info(String.format("total row num: %d", p.getRowNum()));

			LongField failedNumField = new LongField("failed_num");
			failedNumField.setValue(Long.valueOf(p.getFailedNum()));
			this.addDataField(failedNumField);
			logger.info(String.format("failed row num: %d", p.getFailedNum()));
			if (p.getFailedNum() > 0) {
				logger.error("cleaning have errors, please check error log");
			}
		} catch (Exception e) {
			e.printStackTrace();
			logger.error(String.format("step[%s] occur error, %s", this.id, e.getMessage()));
			return 1;
		} finally {
			if (p != null) {
				p.close();
			}
		}

		return 0;
	}

}
